package com.katze.common.heart.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

public class HeartBeatClient {
    private static final Logger log = LoggerFactory.getLogger(HeartBeatClient.class);

    private static final long RECONNECT_SECONDS = 5;
    private final EventLoopGroup group = new NioEventLoopGroup(1);
    private Channel channel;
    private final SocketAddress address;
    private Supplier<String> active;

    public HeartBeatClient(SocketAddress address){
        this.address = address;
    }

    public void initialization()  {
        HeartBeatClient client = this;
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.remoteAddress(address);
        bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("decoder", new StringDecoder());
                pipeline.addLast("encoder", new StringEncoder());
                pipeline.addLast(new HeartBeatClientHandler(client));
            }
        });

        bootstrap.connect().addListener((ChannelFutureListener) future -> {
            if (!future.isSuccess()){
                log.warn("Heart Beat Client Connection {} Fail!", address.toString());
                reconnect();
            } else {
                this.channel = future.channel();
                log.info("Heart Beat Client Connection Success!");
                this.channel.writeAndFlush(active.get());
            }
        });
    }

    protected void reconnect() {
        group.schedule(this::initialization, RECONNECT_SECONDS, TimeUnit.SECONDS);
    }

    /**
     * 发送消息
     */
    protected void send(String message) {
        if (this.channel != null && this.channel.isActive()) {
            this.channel.writeAndFlush(message);
        }
    }

    /**
     * 发送心跳数据包
     */
    public void heartbeat() {
        group.scheduleWithFixedDelay(()-> send("Heartbeat"), 5, 150, TimeUnit.SECONDS);
    }

    /**
     * 连接成功后首次发送消息
     */
    public void active(Supplier<String> active){
        this.active = active;
    }

    public void destroy() {
        group.shutdownGracefully();
        log.info("Heart Beat Client Close!");
    }
}
